package com.df.ez_game.server.broker;

import cn.hutool.core.date.DateTime;
import com.iohao.game.common.kit.NetworkKit;
import io.scalecube.cluster.Cluster;
import io.scalecube.cluster.ClusterImpl;
import io.scalecube.cluster.ClusterMessageHandler;
import io.scalecube.cluster.membership.MembershipEvent;
import io.scalecube.cluster.metadata.MetadataCodec;
import io.scalecube.cluster.transport.api.Message;
import io.scalecube.net.Address;
import io.scalecube.transport.netty.tcp.TcpTransportFactory;
import lombok.Builder;
import lombok.Data;
import lombok.experimental.Accessors;
import lombok.extern.slf4j.Slf4j;
import org.jctools.maps.NonBlockingHashMap;

import java.io.Serializable;
import java.util.*;
import java.util.function.Function;

/**
 * 集群节点
 */
@Data
@Accessors(chain = true)
public class ClusterNode {
    private List<String> seedAddress = new ArrayList<>();
    private int port;
    private Cluster cluster;
    private String name;

    private Map<String,ClusterNodeMeta> metas = new HashMap<>();

    /**
     * 启动，加入gossip网络
     */
    public void start() {
        final String localIp = NetworkKit.LOCAL_IP;

        this.name = String.format("ioGameCluster-%d--%s", port,  localIp);
        ClusterNodeMeta clusterNodeMeta = new ClusterNodeMeta();
        clusterNodeMeta.setId(UUID.randomUUID().toString())
                .setArea(0)
                .setPort(port)
                .setStartTime(DateTime.now().getTime())
                .setAddrString(localIp.toString()+":"+port)
                .setAddress(localIp);

        // 种子节点地址
        List<Address> seedMemberAddress = this.listSeedMemberAddress();

        this.cluster = new ClusterImpl()
                .config(options -> options
                        .memberAlias(name)
                        .metadata(clusterNodeMeta)
                        .metadataCodec(new JsonMetaCodec())
                        .externalHost(localIp)
                        // externalPort是一个容器环境的配置属性，它被设置为向 scalecube 集群发布一个映射到 scalecube 传输侦听端口。
                        .externalPort(port)
                )
                // 种子成员地址
                .membership(membershipConfig -> membershipConfig
                        // 种子节点地址
                        .seedMembers(seedMemberAddress)
                        // 时间间隔
                        .syncInterval(1_000)
                )
                .handler(cluster->{
                   return new ClusterMessageHandler(cluster,this);
                })
                .transportFactory(TcpTransportFactory::new)
                .transport(transportConfig -> transportConfig.port(port))
                .startAwait();
        metas.put(clusterNodeMeta.addrString, clusterNodeMeta);

    }

    /**
     * 主动推出gossip网络，断开所有连接的逻辑服连接
     */
    public void shutdown(){

    }

    private List<Address> listSeedMemberAddress() {
        return this.seedAddress
                .stream()
                .map(Address::from)
                .toList();
    }

    @Data
    @Accessors(chain = true)
    public static class ClusterNodeMeta implements Serializable {
        private String address;
        private int port;
        private String id;
        private String addrString;//ip:port
        private long startTime;
        private int area;//地区编号
        //TODO
        //CPU 信息
        //网络流量信息
        //请求数量信息(转发网关得信息数，内网之间得西澳西转发数)
        //还可以有其他信息，让用户自己填充
    }
    @Slf4j
    private static class ClusterMessageHandler implements io.scalecube.cluster.ClusterMessageHandler {
        private Cluster cluster;
        private ClusterNode node;

        public ClusterMessageHandler(Cluster cluster, ClusterNode node) {
            this.cluster = cluster;
            this.node = node;
        }

        @Override
        public void onMessage(Message message) {
            log.info("\n{}",  " received: " + message.data());
        }

        @Override
        public void onGossip(Message gossip) {
            log.info("\n{}",  "gossip received: " + gossip.data());
        }

        @Override
        public void onMembershipEvent(MembershipEvent event) {
            Map<String, ClusterNodeMeta> brokers = new HashMap<>();

            cluster.members().forEach(member -> {
                Optional<ClusterNodeMeta> optional = cluster.metadata(member);
                optional.ifPresent(metadata -> {
                    ClusterNodeMeta clusterNodeMeta = new ClusterNodeMeta();

                    brokers.put(clusterNodeMeta.addrString, clusterNodeMeta);
                });
            });

            // 使用新的 brokers。无论是 ADDED、REMOVED 都重新生成一次。
            this.node.metas = brokers;

            this.notifyServerNode();
        }

        //通知连接此broker的所有服务器节点
        private void notifyServerNode() {

        }
    }
}
